/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.hylanda.nifi.processor.ams.processors.RFMAnalyzeModel;

import com.hylanda.gate.prs.StandardDate;
import net.minidev.json.JSONObject;
import net.minidev.json.JSONValue;
import org.apache.nifi.annotation.behavior.*;
import org.apache.nifi.annotation.documentation.CapabilityDescription;
import org.apache.nifi.annotation.documentation.SeeAlso;
import org.apache.nifi.annotation.documentation.Tags;
import org.apache.nifi.annotation.lifecycle.OnScheduled;
import org.apache.nifi.annotation.lifecycle.OnStopped;
import org.apache.nifi.components.PropertyDescriptor;
import org.apache.nifi.components.state.Scope;
import org.apache.nifi.flowfile.FlowFile;
import org.apache.nifi.processor.*;
import org.apache.nifi.processor.exception.ProcessException;
import org.apache.nifi.processor.util.StandardValidators;
import org.apache.nifi.util.LogAppendInfoUtil;

import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.text.SimpleDateFormat;
import java.util.*;

@CustomDataPreferred
@ShowState(keys = { @StateKey(key = "提示"),@StateKey(key = "注意")})
@Tags({"example"})
@CapabilityDescription("Provide a description")
@SeeAlso({})
@ReadsAttributes({@ReadsAttribute(attribute = "", description = "")})
@WritesAttributes({@WritesAttribute(attribute = "", description = "")})
public class RfmProcessor extends AbstractProcessor {

    public static final PropertyDescriptor FIELD_BUYER_ID_PROPERTY = new PropertyDescriptor
            .Builder().name("FIELD_BUYER_ID_PROPERTY")
            .displayName("字段名称-客户网名")
            .description("设置客户网名在数据中的字段名称，以便系统识别该数据")
            .defaultValue("客户网名")
            .required(true)
            .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
            .build();

    public static final PropertyDescriptor FIELD_ORDER_ID_PROPERTY = new PropertyDescriptor
            .Builder().name("FIELD_ORDER_ID_PROPERTY")
            .displayName("字段名称-订单编号")
            .description("设置订单号在数据中的字段名称，以便系统识别该数据")
            .defaultValue("订单编号")
            .required(true)
            .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
            .build();

    public static final PropertyDescriptor FIELD_ORDER_TIME_PROPERTY = new PropertyDescriptor
            .Builder().name("FIELD_ORDER_TIME_PROPERTY")
            .displayName("字段名称-订单时间")
            .description("设置订单时间在数据中的字段名称，以便系统识别该数据")
            .defaultValue("交易时间")
            .required(true)
            .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
            .build();
    public static final PropertyDescriptor FIELD_ORDER_COST_PROPERTY = new PropertyDescriptor
            .Builder().name("FIELD_ORDER_COST_PROPERTY")
            .displayName("字段名称-订单金额")
            .description("设置订单金额在数据中的字段名称，以便系统识别该数据")
            .defaultValue("订单支付金额")
            .required(true)
            .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
            .build();


//    public static final Relationship MY_RELATIONSHIP = new Relationship.Builder()
//            .name("MY_RELATIONSHIP")
//            .description("Example relationship")
//            .build();

    private List<PropertyDescriptor> descriptors;

    private Set<Relationship> relationships;

    private String _field_buyer_id = null;
    private String _field_order_id = null;
    private String _field_order_time = null;
    private String _field_order_cost = null;
    private DockerHolder _srv = new DockerHolder();

    @Override
    protected void init(final ProcessorInitializationContext context) {
        final List<PropertyDescriptor> descriptors = new ArrayList<PropertyDescriptor>();
        descriptors.add(FIELD_BUYER_ID_PROPERTY);
        descriptors.add(FIELD_ORDER_ID_PROPERTY);
        descriptors.add(FIELD_ORDER_TIME_PROPERTY);
        descriptors.add(FIELD_ORDER_COST_PROPERTY);


        this.descriptors = Collections.unmodifiableList(descriptors);

        final Set<Relationship> relationships = new HashSet<Relationship>();
        //relationships.add(MY_RELATIONSHIP);
        this.relationships = Collections.unmodifiableSet(relationships);
    }

    @Override
    public Set<Relationship> getRelationships() {
        return this.relationships;
    }

    @Override
    public final List<PropertyDescriptor> getSupportedPropertyDescriptors() {
        return descriptors;
    }

    @OnScheduled
    public void onScheduled(final ProcessContext context) throws Exception {
        context.getStateManager().setState(new HashMap<String, String>() {{
            put("提示", "rfm模型正在初始化...");
            put("data_url", "");
        }}, Scope.LOCAL);
        String userId = LogAppendInfoUtil.userId();
        String projectId = LogAppendInfoUtil.projectId();
        _field_buyer_id = context.getProperty("FIELD_BUYER_ID_PROPERTY").getValue();
        _field_order_id = context.getProperty("FIELD_ORDER_ID_PROPERTY").getValue();
        _field_order_time = context.getProperty("FIELD_ORDER_TIME_PROPERTY").getValue();
        _field_order_cost = context.getProperty("FIELD_ORDER_COST_PROPERTY").getValue();
        _srv.init(this.getIdentifier());
        String strSvrPageUrl = _srv.getSvrShowUrlForWlan(userId,projectId);
        if(! _srv.IsInitSuc()){
            context.getStateManager().setState(new HashMap<String, String>() {{
                put("提示", "无法初始化rfm模型");
                put("data_url", "");
            }}, Scope.LOCAL);
            throw new Exception("无法初始化rfm模型");
        }else {
            if (strSvrPageUrl != null && !strSvrPageUrl.isEmpty()) {
                context.getStateManager().setState(new HashMap<String, String>() {{
                    put("提示", "初始化成功，请在右侧打开数据标签");
                    put("data_url", strSvrPageUrl);
                }}, Scope.LOCAL);
            } else {
                context.getStateManager().setState(new HashMap<String, String>() {{
                    put("提示", "申请rfm模型使用授权失败");
                    put("data_url", "");
                }}, Scope.LOCAL);
                throw new Exception("申请rfm模型使用授权失败，请联系能力供应商");
            }
        }
    }

    @OnStopped
    public void onStopped(final ProcessContext context) throws IOException {
        _srv.uninit();
    }


    public static byte[] readStream(InputStream inStream) throws Exception {
        ByteArrayOutputStream outSteam = new ByteArrayOutputStream();
        byte[] buffer = new byte[1024];
        int len = -1;
        while ((len = inStream.read(buffer)) != -1) {
            outSteam.write(buffer, 0, len);
        }
        outSteam.close();
        inStream.close();
        return outSteam.toByteArray();
    }

    @Override
    public void onTrigger(final ProcessContext context, final ProcessSession session) throws ProcessException {
        for (int i = 0; i < 100; i++) {
            FlowFile flowFile = session.get();
            if (flowFile == null) {
                return;
            }
            //读取FlowFile
            String content = "";
            InputStream in = session.read(flowFile);
            try {
                byte[] by = readStream(in);
                content = new String(by, "UTF-8");
                try {
                    JSONObject jsonInput = (JSONObject) JSONValue.parse(content);
                    String buyer_id = jsonInput.getAsString(_field_buyer_id);
                    String order_id = jsonInput.getAsString(_field_order_id);
                    String order_time = jsonInput.getAsString(_field_order_time);
                    String order_cost = jsonInput.getAsString(_field_order_cost);
                    if (buyer_id != null && !buyer_id.isEmpty()
                            && order_id != null && !order_id.isEmpty()
                            && order_time != null && !order_time.isEmpty()
                            && order_cost != null && !order_cost.isEmpty()
                            ) {
                        Date order_date = StandardDate.GetStandardDate(order_time);
                        if (order_date != null) {
                            order_time = (new SimpleDateFormat("yyyy-MM-dd HH:mm:ss")).format(order_date);
                            long start = System.currentTimeMillis();
                            String strRet = _srv.calc("RFM.input", new Object[]{order_id, buyer_id, order_time, order_cost});
                            long time = System.currentTimeMillis() - start;
                            System.out.print(strRet+" ");
                            System.out.println(time);
                        }
                    }
                    session.remove(flowFile);

                } catch (Throwable throwable) {
                    throwable.printStackTrace();
                }
            } catch (Exception e) {
                // TODO Auto-generated catch block
                e.printStackTrace();
            }

        }
    }
}
